package com.qf.work;

import com.qf.util.MQConnUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

    @Test
    public void consume1() throws Exception {
//        - 获取Connection
        Connection conn = MQConnUtil.getConn();
//        - 创建Channel
        Channel channel = conn.createChannel();
//        - 创建Queue
        channel.queueDeclare(Publisher.QUEUE_NAME,true,false,false,null);

        channel.basicQos(2);
//        - 监听Queue中的消息
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者1号获取到消息：" + new String(body));
                // 手动ack
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // autoAck：设置为true,代表自动ack,只要拿到消息,立即告诉MQ,消息处理完了.
        // autoAck：应当设置为false，代表手动ack，在执行完毕业务代码后，再给MQ发送ack
        channel.basicConsume(Publisher.QUEUE_NAME,false,callback);
//        - 让程序停在这个位置
        System.out.println("消费者1号开始消费消息!!");
        System.in.read();
//        - 释放资源
        channel.close();
        conn.close();
    }

    @Test
    public void consume2() throws Exception, TimeoutException {
//        - 获取Connection
        Connection conn = MQConnUtil.getConn();
//        - 创建Channel
        Channel channel = conn.createChannel();
//        - 创建Queue
        channel.queueDeclare(Publisher.QUEUE_NAME,true,false,false,null);

        channel.basicQos(2);
//        - 监听Queue中的消息
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2号获取到消息：" + new String(body));
                // 手动ack
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,false,callback);
//        - 让程序停在这个位置
        System.out.println("消费者2号开始消费消息!!");
        System.in.read();
//        - 释放资源
        channel.close();
        conn.close();
    }

}
